package com.bigfans.framework.kafka;

import com.bigfans.framework.utils.JsonUtils;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaTemplate {

    private KafkaFactory connectionFactory;

    public KafkaTemplate(KafkaFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    public <T> void send(String topic , T message){
        Producer producer = connectionFactory.getProducer();
        producer.send(new ProducerRecord(topic , JsonUtils.toJsonString(message)));
    }

    public <T> void send(T event){
        this.send(event.getClass().getName() , event);
    }

}
